package com.codejiwei.flink.practice;

import com.codejiwei.flink.entity.OrderEvent;
import com.codejiwei.flink.entity.TxEvent;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @author jiwei
 * @description
 * @date 2023/5/23 10:23
 */
public class Flink_Project_Order {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.readTextFile("src/main/resources/data/OrderLog.csv")
                .map(line -> {
                    String[] fields = line.split(",");
                    return new OrderEvent(
                            Long.valueOf(fields[0]),
                            fields[1],
                            fields[2],
                            Long.valueOf(fields[3])
                    );
                });
        SingleOutputStreamOperator<TxEvent> txEventStream = env.readTextFile("src/main/resources/data/ReceiptLog.csv")
                .map(line -> {
                    String[] fields = line.split(",");
                    return new TxEvent(
                            fields[0],
                            fields[1],
                            Long.valueOf(fields[2])
                    );
                });
        ConnectedStreams<OrderEvent, TxEvent> connectedStreams = orderEventStream.connect(txEventStream);

        connectedStreams.keyBy("txId", "txId")
                .process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
                    // 存 txId -> OrderEvent
                    Map<String, OrderEvent> orderMap = new HashMap<>();
                    // 存储 txId -> TxEvent
                    Map<String, TxEvent> txMap = new HashMap<>();
                    @Override
                    public void processElement1(OrderEvent orderEvent, Context context, Collector<String> collector) throws Exception {
                        if (txMap.containsKey(orderEvent.getTxId())) {
                            collector.collect("订单: " + orderEvent + "对账成功");
                            txMap.remove(orderEvent.getTxId());
                        } else {
                            orderMap.put(orderEvent.getTxId(), orderEvent);
                        }
                    }

                    @Override
                    public void processElement2(TxEvent txEvent, Context context, Collector<String> collector) throws Exception {
                        if (orderMap.containsKey(txEvent.getTxId())) {
                            OrderEvent orderEvent = orderMap.get(txEvent.getTxId());
                            collector.collect("订单: " + orderEvent + "对账成功");
                            orderMap.remove(txEvent.getTxId());
                        } else {
                            txMap.put(txEvent.getTxId(), txEvent);
                        }

                    }
                })
                .print();

        env.execute();


    }
}
